package com.ideal.lx_mss.utils;




import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.ideal.lx_mss.common.MssLogs;
import com.ideal.lx_mss.common.TableInfo;
import com.ideal.lx_mss.entity.KafkaTopicTableEntity;
import com.ideal.lx_mss.entity.PullFileEntity;
import com.ideal.lx_mss.entity.PullInfoEntity;
import com.ideal.lx_mss.mapper.BasicMapper;
import com.ideal.lx_mss.mapper.KafkaTopicTableMapper;
import com.ideal.lx_mss.mapper.MssLogsMapper;
import com.ideal.lx_mss.mapper.PullFileMapper;
import com.ideal.lx_mss.service.DataService;

import org.json.JSONException;
import org.json.JSONObject;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.BadSqlGrammarException;
import org.springframework.stereotype.Component;

import java.io.*;
import java.util.*;
import java.util.concurrent.*;

/**
 * 分析工具
 */
@Component
public class AnalysisUtils {

    @Autowired
    private DataService dataService;
    @Autowired
    private CsvUtils csvUtils;

    @Autowired
    private TextUtils textUtils;
    @Autowired
    private MssLogsMapper mssLogsMapper;
    @Autowired
    private SqlUtils sqlUtils;
    @Autowired
    private BasicMapper basicMapper;

    @Autowired
    private PullFileMapper pullFileMapper;
    @Autowired
    private KafkaTopicTableMapper kafkaTopicTableMapper;

    private List<String> currentBatch = new ArrayList<>();




    public boolean structureAnalysisByFile(String formId,String fileUrl,String topic){


        //查询topic对应的目标表
        QueryWrapper<KafkaTopicTableEntity> kafkaTopicTableEntityQueryWrapper = new QueryWrapper<>();
        kafkaTopicTableEntityQueryWrapper.eq("topic_name",topic);
        List<KafkaTopicTableEntity> kafkaTopicTableEntities = kafkaTopicTableMapper.selectList(kafkaTopicTableEntityQueryWrapper);

        for (KafkaTopicTableEntity kafkaTopicTableEntity : kafkaTopicTableEntities) {
            basicMapper.truncateTables(kafkaTopicTableEntity.getTempTableName());
        }



        HashSet<String> set = new HashSet<>();
        //读文件
        File folder = new File(fileUrl);

        if (folder.isDirectory()) {
            File[] files = folder.listFiles();
            if (files != null) {
                for (File file : files) {
                    try (BufferedReader reader = new BufferedReader(new FileReader(file))) {
                        String line;
                        while ((line = reader.readLine()) != null) {
                            JSONObject jsonObject = new JSONObject(line);

                            String itfcode = jsonObject.getString("itfcode");//表名
                            String datatype = jsonObject.getString("datatype");
                            jsonObject.put("kafkaMgsType",datatype);
                            try {
                                insertMssTemp(kafkaTopicTableEntities,jsonObject,topic);
                            }catch (Exception e){
                                //异常数据
                                currentBatch.add(line);
                                set.add(itfcode);
                            }


                        }


                    } catch (IOException | JSONException e) {
                        e.printStackTrace();
                    }


                }

            }
        }
        if (!set.isEmpty()){
            //本次数据操作涉及到的表名
            basicMapper.insertTableNameTemp(set,formId);
        }


        if (currentBatch.size()>0){
            fileUrl = fileUrl+"/err";
            textUtils.saveText(currentBatch,topic,formId,fileUrl,"err");
            PullFileEntity pullFileEntity = new PullFileEntity(UUID.randomUUID().toString(), formId, fileUrl, CommonUtil.getNowDate(), topic, "0","err");
            pullFileMapper.insert(pullFileEntity);
        }


        return true;
    }

    public void insertMssTemp(List<KafkaTopicTableEntity> kafkaTopicTableEntities,JSONObject jsonObject,String topic) throws Exception{
        String itfcode = jsonObject.getString("itfcode");//表名
        JSONObject dataload = jsonObject.getJSONObject("dataload");//数据
        try {
            for (KafkaTopicTableEntity kafkaTopicTableEntity : kafkaTopicTableEntities) {
                if (itfcode.equals(kafkaTopicTableEntity.getTableName())){
                    String insertSql = sqlUtils.generateInsertSQL(kafkaTopicTableEntity.getTempTableName(), dataload);
                    basicMapper.insertTableUtilsMapper(insertSql);
                    break;
                }
            }
        }catch (Exception e){
            throw e;
        }






    }




}
